package com.ydl.learning.flink.demo.pipeline

import com.ydl.learning.flink.demo.pipeline.Pipeline.Builder
import com.ydl.learning.flink.demo.pipeline.java.entity.JobConfig
import org.apache.flink.streaming.api.scala.DataStream


/**
 *
 *
 * @author ydl
 * @since 2022/1/7
 */
class Pipeline[T] {
  private var jobConfig: JobConfig = _
  private var init: Source[T] = _
  private var processes: List[Process[T]] = _
  private var sink: Sink[T] = _

  def this(builder: Builder[T]) = {
    this
    this.jobConfig = builder.jobConfig
    this.init = builder.init
    this.processes = builder.processes
    this.sink = builder.sink

  }

  def execute(): Unit = {
    init.setConfig(this.jobConfig)
    init.checkConfig()
    var result: DataStream[T] = init.init()
    for (process <- processes) {
      process.setData(result)
      result = process.run
    }
    sink.setSource(result)
    sink.sink()
    init.env().execute(jobConfig.getJobName)
  }


}

object Pipeline {

  class Builder[S] {
    private[this] var _jobConfig: JobConfig = null

    def jobConfig: JobConfig = _jobConfig

    private[this] def jobConfig_=(value: JobConfig): Unit = {
      _jobConfig = value
    }

    private[this] var _init: Source[S] = null

    def init: Source[S] = _init

    private[this] def init_=(value: Source[S]): Unit = {
      _init = value
    }

    private[this] var _processes: List[Process[S]] = List[Process[S]]()

    def processes: List[Process[S]] = _processes

    private[this] def processes_=(value: List[Process[S]]): Unit = {
      _processes = value
    }


    //TODO 测流输出
    private var processSideOut:String = null

    //TODO 监控
    private var processMonitor = null

    private[this] var _sink: Sink[S] = null

    def sink: Sink[S] = _sink

    private[this] def sink_=(value: Sink[S]): Unit = {
      _sink = value
    }

    //TODO 流克隆
    private var cloneStream = null
    //TODO 克隆流处理
    private var cloneStreamProcess = null
    //TODO 克隆流输出
    private var cloneStreamSink = null

    def jobConfig(jobConfig: JobConfig): Builder[S] = {
      this.jobConfig = jobConfig
      this
    }

    def init(init: Source[S]): Builder[S] = {
      this.init = init
      this
    }

    def process(process: Process[S]): Builder[S] = {
      this.processes = this.processes :+ process
      this
    }

    def sink(sink: Sink[S]): Builder[S] = {
      this.sink = sink
      this
    }

    def build = new Pipeline[S](this)

  }

}
